package com.ruyuan.ingestion.kafka.streaming;
import com.ruyuan.ingestion.kafka.BaseStreamer;
import java.util.ArrayList;
import java.util.List;

public class EventAttendeesStreamer extends BaseStreamer {

    @Override
    protected String getApplicationId() {
        return "eventAttendees"+getTargetTopic();
    }

    @Override
    protected String getSourceTopic() {
        return "event_attendees_raw";
    }

    @Override
    protected String getTargetTopic() {
        return "event_attendees";
    }

    /**
     * fields：1159822043,1975964455 252302513 4226086795 3805886383,1420484491 3831921392 3973364512,2733420590 517546982 1350834692 532087573 583146976 3079807774 1324909047,1723091036 3795873583 4109144917 3560622906 3106484834 2925436522 2284506787
     *  event_id 1159822043
     *  yes:1975964455 252302513 4226086795 3805886383
     *  maybe:1420484491 3831921392 3973364512
     *  ....
     *
     * */
    @Override
    protected List<String[]> transform(String[] fields) {
        List<String[]> result = new ArrayList<>();
        String evnet_id = fields[0];

        //yes的状态
        if (fields.length > 1 && fields[1] != null) {
            String[] user_yes = fields[1].split(" ");
            if (user_yes !=null && user_yes.length > 0) {
                for(String yes : user_yes) {
                    result.add(new String[]{evnet_id,yes,"yes"});
                }
            }
        }

        //maybe
        if (fields.length > 2 && fields[2] != null) {
            String[] maybes = fields[2].split(" ");
            if (maybes != null && maybes.length>0) {
                for(String maybe:maybes) {
                    result.add(new String[]{evnet_id,maybe,"maybe"});
                }
            }
        }

        // invited
        if (fields.length > 3 && fields[3] != null) {
            String[] inviteds = fields[3].split(" ");
            if (inviteds != null && inviteds.length>0) {
                for(String invited:inviteds) {
                    result.add(new String[]{evnet_id,invited,"invited"});
                }
            }
        }

        // no
        if (fields.length > 4 && fields[4] != null) {
            String[] noUsers = fields[4].split(" ");
            if (noUsers != null && noUsers.length>0) {
                for(String no:noUsers) {
                    result.add(new String[]{evnet_id,no,"no"});
                }
            }
        }
        return result;
    }

    @Override
    protected Boolean isHeader(String[] fields) {
        return (isValid(fields) && fields[0].equals("event") && fields[1].equals("yes") && fields[2].equals("maybe") && fields[3].equals("invited") && fields[4].equals("no"));
    }

    @Override
    protected Boolean isValid(String[] fields) {
        return (fields.length == 5);
    }
}
